package com.lp.canal;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.Message;
import com.lp.enums.CanalFactoryEnum;
import com.lp.util.StringConvertUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author 10263
 */
@Component
@Slf4j
public class CanalHandler {

    @Resource
    private CanalProperties canalProperties;

    /**
     * 数据渲染，调用业务方法
     *
     * @param entrys
     */
    private static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            try {
                RowChange rowChage = RowChange.parseFrom(entry.getStoreValue());
                EntryHandler entryHandler = SpringUtil.getBean(CanalFactoryEnum.getClazz(entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName()));
                if (entryHandler != null) {
                    //获取具体类型
                    Type genericSuperclass = entryHandler.getClass().getGenericInterfaces()[0];
                    List<Type> list = null;
                    if (genericSuperclass instanceof ParameterizedType parameterizedType) {
                        Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
                        list = Arrays.asList(actualTypeArguments);
                    }
                    if (CollUtil.isNotEmpty(list)) {
                        String aClass = list.get(0).getTypeName();
                        EventType eventType = rowChage.getEventType();
                        for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                            if (eventType == EventType.DELETE) {
                                entryHandler.delete(newInstance(Class.forName(aClass), rowData.getBeforeColumnsList()));
                            } else if (eventType == EventType.INSERT) {
                                entryHandler.insert(newInstance(Class.forName(aClass), rowData.getAfterColumnsList()));
                            } else {
                                entryHandler.update(newInstance(Class.forName(aClass), rowData.getBeforeColumnsList()), newInstance(Class.forName(aClass), rowData.getAfterColumnsList()));
                            }
                        }
                    }
                }
            } catch (Exception e) {
                log.error("ERROR ## parser of eromanga-event has an error , data:" + entry, e);
            }
        }
    }

    /**
     * 实例化
     *
     * @param c
     * @param columns
     * @param <T>
     * @return
     */
    private static <T> T newInstance(Class<T> c, List<CanalEntry.Column> columns) {
        try {
            T obj = c.getDeclaredConstructor().newInstance();
            Map<String, Field> columnNames = Arrays.stream(obj.getClass().getDeclaredFields()).collect(Collectors.toMap(Field::getName, Function.identity()));
            for (CanalEntry.Column column : columns) {
                Field field = columnNames.get(toCamelCase(column.getName()));
                if (field != null) {
                    field.setAccessible(true);
                    Class<?> type = field.getType();
                    Object result = StringConvertUtil.convertType(type, column.getValue());
                    field.set(obj, result);
                }
            }
            return obj;
        } catch (InstantiationException | IllegalAccessException | NoSuchMethodException |
                 InvocationTargetException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 驼峰式命名法
     */
    private static String toCamelCase(String s) {
        if (s == null) {
            return null;
        }
        s = s.toLowerCase();
        StringBuilder sb = new StringBuilder(s.length());
        boolean upperCase = false;
        for (int i = 0; i < s.length(); i++) {
            char c = s.charAt(i);
            if (c == '_') {
                upperCase = true;
            } else if (upperCase) {
                sb.append(Character.toUpperCase(c));
                upperCase = false;
            } else {
                sb.append(c);
            }
        }
        return sb.toString();
    }

    /**
     * 开始监听canal
     */
    @Async
    public void initialize() {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(this.canalProperties.getHostname(),
                this.canalProperties.getPort()), this.canalProperties.getDestination(), this.canalProperties.getUsername(), this.canalProperties.getPassword());
        try {
            connector.connect();
            connector.subscribe(CanalFactoryEnum.getTables());
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(1);
                long batchId = message.getId();
                if (batchId != -1L && message.getEntries().size() != 0) {
                    printEntry(message.getEntries());
                }
                connector.ack(batchId);
            }
        } catch (Exception e) {
            log.error("canal client 异常", e);
        } finally {
            connector.disconnect();
        }
    }


}